package com.hy.main.demo.rabbitmq;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.MessageProperties;

import java.util.Scanner;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * @Author yang DongYong
 * @Date 2022/5/9 - 05 - 09 - 0:10
 * @Description: com.hy.main.demo.rabbitmq
 * @version: 1.0
 */
public class Task02 {
    private static final String QUEUE_NAME = "ack_queue";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        // 开启发布确认
        channel.confirmSelect();
        // 持久化队列 队列如果第一次创建可以直接持久化 如果已经存在的队列则需要先删除
        boolean durable = true;
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, durable, false, false, null);

        // 记录消息 要使用线程安全的调表 并且支持高并发和多线程
        ConcurrentSkipListMap<Long, String> messagesMap = new ConcurrentSkipListMap<>();

        // 消息确认成功
        ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
            System.out.println(multiple);
            // 批量发送
            if (multiple) {
                // 删除已经确认的消息剩下的就是还未确认的消息
                ConcurrentNavigableMap<Long, String> confirmedMap = messagesMap.headMap(deliveryTag);
                channel.clearReturnListeners();
            } else {
                // 单次确认
                messagesMap.remove(deliveryTag);
            }


            System.out.println("确认的消息:" + deliveryTag);
        };
        // 未确认的消息
        ConfirmCallback nackCallback = (deliveryTag, multiple) -> {
            // 未确认的消息
            String nackMessage = messagesMap.get(deliveryTag);
            System.out.println("未确认的消息id:"+nackMessage +":::::::"+"未确认的消息:" + deliveryTag);
        };

        // 准备消息的监听器 第一个参数是已经确认的消息内容，第二个参数是未确认的消息内容
        channel.addConfirmListener(ackCallback, nackCallback);
        Scanner scanner = new Scanner(System.in);
        while (scanner.hasNext()) {
            String message = scanner.next();
            /**
             * 发送一个消费者
             * 1、发送到那个交换机
             * 2、路由的key是哪一个 本次是队列的名称
             * 3、其他参数信息
             * 4、要发送的具体消息内容
             */
            // 设置生产者发送的消息为持久化
            channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes("UTF-8"));
            System.out.println("生产者发送的消息为:" + message);
            // 发布确认
            boolean b = channel.waitForConfirms();
            // 此处记录下所有发送的消息
            messagesMap.put(channel.getNextPublishSeqNo(), message);
        }
    }
}
